package mq

import (
	"fmt"
	"time"
	"pd/server/plugin/cache"

	"github.com/google/uuid"
)

type redisMQ struct {
	retryTime time.Duration // 没有删除的消息，等待retryTime时间后会重新被接收
	topic     string
	redis     *cache.RedisClient
}

type RedisMsgNode struct {
	ID      string `json:"id"`
	Content string `json:"c"`
}

func (m *redisMQ) Subscribe(topic string) error {
	m.topic = topic
	return nil
}

func (m *redisMQ) getTopic() string {
	return "rmq_" + m.topic
}
func (m *redisMQ) getDataKey() string {
	return "rmqc_" + m.topic
}

// 接收消息(阻塞模式)
func (m *redisMQ) Receive(timeout time.Duration) (ok bool, msg *string, msgID *string, err error) {
	var waitTime time.Duration = 0
	step := time.Second
	var ts *float64
	var idStr string
	for {
		now := time.Now()
		ts, idStr, err = m.redis.ZTopWithScores(m.getTopic())
		if err != nil {
			return false, nil, nil, err
		}
		if ts != nil && *ts <= float64(now.UnixNano()) { // 有消息
			break
		}

		waitTime += step
		if waitTime >= timeout {
			return false, nil, nil, nil
		}
		time.Sleep(step)
	}

	// 从hmap中取出消息内容
	content, err := m.redis.C().HGet(m.getDataKey(), idStr).Result()
	if err != nil {
		return
	}

	// 设置超时时间，超过该时间后如果消息没有被删除，则将会再次被接收
	if _, err = m.redis.C().ZIncrBy(m.getTopic(), float64(m.retryTime), idStr).Result(); err != nil {
		return
	}

	ok = true
	msg = &content
	msgID = &idStr

	return
}

// 发布消息
const (
	PushScript = `
	redis.call('zadd', KEYS[1], KEYS[2], KEYS[3])
	redis.call('hmset', KEYS[4], KEYS[3], KEYS[5])
	return 'OK'
	`
)

func (m *redisMQ) Push(msg *string, delay ...time.Duration) error {
	now := time.Now()
	msgID := uuid.New().String()
	t := now.UnixNano()
	if len(delay) > 0 {
		t = now.Add(delay[0]).UnixNano()
	}
	_, err := m.redis.C().Eval(PushScript, []string{m.getTopic(), fmt.Sprintf("%d", t), msgID, m.getDataKey(), *msg}).Result()

	return err
}

// 删除消息
const deleteScript = `redis.call('zrem', KEYS[1], KEYS[2]) redis.call('hdel', KEYS[3], KEYS[2]) return 'OK'`

func (m *redisMQ) Delete(msgID *string) error {
	_, err := m.redis.C().Eval(deleteScript, []string{m.getTopic(), *msgID, m.getDataKey()}).Result()
	return err
}
